package com.atguigu.sink;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/15 9:47
 *
 * Kafka Connector:
 *      1.KafkaSource
 *      2.KafkaSink
 *          生产者：KafkaProducer
 *          生产者配置：
 *              集群位置：bootstrap.Service
 *              key的序列化器：key.serializer
 *              value的序列化器：value.serializer
 *              缓冲区大小：buffer.memory（默认32M）
 *              每个分区缓冲区大小（批次大小）：batch.size
 *              每批次超时时间： linger.ms
 *              应答级别： acks
 *              事务超时时间：transaction。timeout。ms
 *              事务ID：transaction。ID
 *          生产者分区分配策略：
 *              默认使用粘性分区
 *                  1. 如果明确指定分区号，直接使用
 *                  2. 如果没有指定分区号，但是指定了key，按照key的hash值计算分区号
 *                  3.如果没有指定分区号和key，直接使用粘性策略
 *
 *          Kafka提供的生产者配置类：ProducerConfig
 */
public class Flink02_KafkaSink {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.enableCheckpointing(5000);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        // 将流中的数据写入到Kafka中
        KafkaSink<Event> kafkaSink = KafkaSink.<Event>builder()
                .setBootstrapServers("hadoop102:9092,hadoop103:9092")
                .setRecordSerializer(
                        new KafkaRecordSerializationSchema<Event>() {
                            /**
                             *ProducerRecord:生产者消息
                             */
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(Event element, KafkaSinkContext context, Long timestap) {
                                ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>("topicA", element.getUser().getBytes(), element.toString().getBytes());
                                return producerRecord;
                            }
                        }
                )
                //EOS(exactly Once semantic):精确一次
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                // 如果将来使用精确一次，必须设置事务ID的前缀
                .setTransactionalIdPrefix("flink" + RandomUtils.nextInt(1, 100))
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,"36000")
                // 如果还有其他配置，可以直接使用setProperty
                //.setKafkaProducerConfig(ProducerConfig.ACKS_CONFIG,"-1")
                .build();

        // 新的API
        ds.sinkTo(kafkaSink);

        env.execute();
    }
}
